package com.bishe.cyh.spark;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * @Author: Poppin
 * @Date: 2022/2/28 17:28
 * @Version: 1.0
 */
public class HBaseClient {

    public static Configuration configuration;
    public static Connection connection;
    public static Admin admin;
    public static Table tableData;
    public static Table tableRe;
    public static Table tableNN;

    public static void init(String tableName){
        configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.quorum","192.168.66.157");
        try {
            connection = ConnectionFactory.createConnection(configuration);
            admin = connection.getAdmin();
            boolean ifexists1 = admin.tableExists(TableName.valueOf(tableName+"data"));
            boolean ifexists2 = admin.tableExists(TableName.valueOf(tableName+"re"));
            boolean ifexists3 = admin.tableExists(TableName.valueOf(tableName+"NN"));
            if (!ifexists1){
                List<ColumnFamilyDescriptor> familyDescriptors = new ArrayList<>();
                familyDescriptors.add(ColumnFamilyDescriptorBuilder.newBuilder("Data".getBytes()).build());
                TableDescriptor descriptorBuilder = TableDescriptorBuilder.newBuilder(TableName.valueOf(tableName+"data")).setColumnFamilies(familyDescriptors).build();
                admin.createTable(descriptorBuilder);
            }
            if (!ifexists2){
                List<ColumnFamilyDescriptor> familyDescriptors = new ArrayList<>();
                familyDescriptors.add(ColumnFamilyDescriptorBuilder.newBuilder("IS".getBytes()).build());
                TableDescriptor descriptorBuilder = TableDescriptorBuilder.newBuilder(TableName.valueOf(tableName+"re")).setColumnFamilies(familyDescriptors).build();
                admin.createTable(descriptorBuilder);
            }
            if (!ifexists3){
                List<ColumnFamilyDescriptor> familyDescriptors = new ArrayList<>();
                familyDescriptors.add(ColumnFamilyDescriptorBuilder.newBuilder("RE".getBytes()).build());
                TableDescriptor descriptorBuilder = TableDescriptorBuilder.newBuilder(TableName.valueOf(tableName+"NN")).setColumnFamilies(familyDescriptors).build();
                admin.createTable(descriptorBuilder);
            }
            tableData = connection.getTable(TableName.valueOf(tableName+"data"));
            tableRe = connection.getTable(TableName.valueOf(tableName+"re"));
            tableNN = connection.getTable(TableName.valueOf(tableName+"NN"));
        }catch (IOException e){
            e.printStackTrace();
        }
    }
    public static void close(){
        try {
            if (admin!=null){
                tableData.close();
                tableNN.close();
                tableRe.close();
                admin.close();
            }
            if (connection!=null){
                connection.close();
            }
        }catch (IOException e){
            e.printStackTrace();
        }
    }

    /**
     * 插入特征码进入数据库
     */
    public static void addRecord(String rowKey,byte[] sift){
        try {
            Put put = new Put(rowKey.getBytes());
            put.addColumn("IS".getBytes(),"sift".getBytes(),sift);
            tableRe.put(put);
        }
        catch (Exception e){
            e.printStackTrace();
        }
    }

    /**
     * 提取全库的特征值
     * @return
     */
    public static List<FaceEnity> FaceAll(){

        try {
            List<FaceEnity> list = new ArrayList<FaceEnity>();
            Scan scan = new Scan();
            ResultScanner scanner = tableRe.getScanner(scan);
            for (Result result:scanner){
                Cell[] cells = result.rawCells();
                for (Cell cell:cells){
                    FaceEnity faceEnity = new FaceEnity();
                    faceEnity.setName(new String(CellUtil.cloneRow(cell)));
                    faceEnity.setFaceFeature(CellUtil.cloneValue(cell));
                    list.add(faceEnity);
                }
            }
            return list;
        }
        catch (Exception e){
            e.printStackTrace();
            return null;
        }

    }

    public static void addData(String rowKey,int age,int gender){

        String Age = Integer.toString(age);
        String Gender = Integer.toString(gender);
        try{
            Put put = new Put(rowKey.getBytes());
            put.addColumn("Data".getBytes(),"age".getBytes(), Bytes.toBytes(Age));
            put.addColumn("Data".getBytes(),"gender".getBytes(),Bytes.toBytes(Gender));
            tableData.put(put);
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    public static void add(String rowKey,String status){
        try {
            Put put = new Put(rowKey.getBytes());
            put.addColumn("RE".getBytes(),"status".getBytes(),status.getBytes());
            tableNN.put(put);
        }catch (Exception e){
            e.printStackTrace();
        }
    }


    public static void dropTable(String tableName){
        try {
            admin.disableTable(TableName.valueOf(tableName));
            admin.deleteTable(TableName.valueOf(tableName));
        }catch (MasterNotRunningException e){
            e.printStackTrace();
        }catch (ZooKeeperConnectionException e){
            e.printStackTrace();
        }catch (IOException e){
            e.printStackTrace();
        }
    }

    public static void deleteTable(String tablename){
        configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.quorum","192.168.66.157");
        try {
            connection = ConnectionFactory.createConnection(configuration);
            admin = connection.getAdmin();
            dropTable(tablename+"data");
            dropTable(tablename+"re");
            dropTable(tablename+"NN");
            close();
        }
        catch (IOException e){
            e.printStackTrace();
        }
    }

}
